
#include <algorithm>
#include <sstream>

#include "chordprotocol.h"
#include "common.h"
#include "chordnode.h"
#include "undertopomgr.h"
#include "chordgod.h"
#include "simulator.h"
#include "randomrange.h"
#include "peerloger.h"
#include "chordkeyinfo.h"
#include "config.h"

uint32_t ChordMsg::msg_id_counter = 0;

FingerItem& FingerItem::operator = ( const ChordNodeInfo& other ){
	node_id_ = other.node_id_;
	chord_nid_ = other.chord_nid_;
	last_update_ = Simulator::instance().clock();
	return *this;
}

void FingerItem::ensure(chord_net_id_t net_id)
{
	if ( isValid() ){
		Node *n = UnderTopoMgr::instance().node( node_id_ );
		bool b_reset = false;
		if ( (n==NULL) ) b_reset = true;
		else{
			ChordNode *cn = static_cast<ChordNode*>(n);
			if ( !(cn->isInNetwork( net_id )) ) b_reset = true;
		}

		if ( b_reset )
			reset();
	}
}

ChordProtocol::ChordProtocol( ChordNode *node, chord_net_id_t net_id )
	: node_(node)
	, net_id_(net_id)
	, number_of_key_(0)
	, status_(CP_ILLEGAL)
	, is_merging_(false)
	, old_net_id_(invalid_chord_net_id)
	, new_net_id_(invalid_chord_net_id)
	, notify_(false)
	, first_node_start_(false)
{
	fingers_.resize( determineFingerTableSize() );
	{//calculate key range for each finger [start_,end_)
		TFingerVector::iterator itr = fingers_.begin(),
			itre = fingers_.end();
		for (int i=0; itr!=itre; itr++,i++ ){
			fingerRange( node_->getChordNodeID(), i, (*itr).start_, (*itr).end_ );
		}
	}

	NEW_OBJECT( RoutineTimer, rt_timer_, (this) );
	NEW_OBJECT( KeyCheckTimer, key_timer_, (this) );
}

ChordProtocol::~ChordProtocol()
{
	RELEASE_OBJ( rt_timer_ );
	RELEASE_OBJ( key_timer_ );

	{
		TMsgItemList::iterator itr = pending_msg_.begin(),
			itre = pending_msg_.end();
		for ( ; itr!=itre; itr++ ){
			MsgItem& item = *itr;
			RELEASE_OBJ( item.msg_ );
		}
		pending_msg_.clear();
	}
	std::for_each( key_list_.begin(), key_list_.end(), ReleaseObject() );
}

/** 
 * @brief Join the Chord network
 * 
 * @return  true if successful
 */
bool ChordProtocol::join()
{
	ChordNode *n = ChordGod::instance().getNodeInChordNet( net_id_, node_->getNodeID() );
	if ( isMerging() ){
		assert( key_list_.empty() );
		ChordProtocol *p_old = node_->getProtocolByID( getOldNetID() );
		assert_or_retv( NULL!=p_old, false );

		key_list_ = p_old->key_list_;
		std::for_each( key_list_.begin(), key_list_.end(), AcquireObject() );
		number_of_key_ = p_old->number_of_key_;

	}
	if ( NULL==n ){ //no node exists
		ChordGod::instance().registerToChordNet( net_id_, node_->getNodeID() );
		setReady();
		getKeysFromGod();
		first_node_start_ = true;
		return true;
	}
	//send a search message to n to find my successor
	chord_key_t key = getFinger(0).start_;
	NEW_VAR_OBJECT( ChordSearchMsg, msg, ( key, ChordMsg::TARGET_SUCCESSOR, getChordNetID(), node_->getNodeID() ) );
	node_->send( n->getNodeID(), msg );
	addPendingMsg( msg );
	setStatus( CP_JOINING );
	RELEASE_OBJ( msg );
	return false;
}

void ChordProtocol::leave()
{
	std::string str;
	keyString( str );
	PeerLoger::instance().logDebug( 0, "node:%u@%u leaves chordnet:%u %s", node_->getChordNodeID(), node_->getNodeID(), net_id_, str.c_str() );
	ChordGod::instance().unregisterToChordNet( net_id_, node_->getNodeID() );
	unregisterKeys();
}

void ChordProtocol::unregisterKeys()
{
	TChordKeyInfoList::iterator itr = key_list_.begin(),
		itre = key_list_.end();
	for ( ; itr!=itre; itr++ ){
		ChordGod::instance().unregisterKeyToChordNet( net_id_, (*itr)->getKey() );
	}
}

void ChordProtocol::routineJob()
{
	if ( !( node_->isOnline() && isActive()) ) return;

	//stabilize
	ChordNodeInfo& suc_info = getSuccessor();
	if( suc_info.isValid() )
	{
		NEW_VAR_OBJECT( ChordGetMsg, get_msg, (ChordMsg::TARGET_PREDCESSOR, net_id_, node_->getNodeID()) );
		node_->send( suc_info.node_id_, get_msg );
		addPendingMsg( get_msg );
		RELEASE_OBJ( get_msg );
	} 
	
	checkDeadFingers();
	fixFingers();
	notify();

	rt_timer_->resched( config.chordprot_.stabilize_intv );
}

void ChordProtocol::fixFingers()
{
	uint32_t count = 0, max_count = 32, index = 0;
	RandomRange range( 1, (uint32_t)fingers_.size() );
	do {
		ChordNode *n = NULL;
		chord_key_t key = getFinger(index).start_;
		//FingerItem& info = closestPrecedingFinger( key );
		//info.ensure();
		//if ( info.isValid() ){
			//n = static_cast<ChordNode*>(UnderTopoMgr::instance().node( info.node_id_ ) );
		//}
		if (NULL==n)
			n = ChordGod::instance().getNodeInChordNet( net_id_, node_->getNodeID() );
		if ( NULL==n ) break; //no node exists

		NEW_VAR_OBJECT( ChordSearchMsg, search_msg, ( key, ChordMsg::TARGET_SUCCESSOR, getChordNetID(), node_->getNodeID() ) );
		node_->send( n->getNodeID(), search_msg );
		addPendingMsg( search_msg );
		RELEASE_OBJ( search_msg );

		count++;
		//assert( range.hasNext() );//sequence cannot be exausted here
		//index = range.next();
		index ++;

	} while( count<max_count );

}

void ChordProtocol::checkDeadFingers()
{
	predecessor_.ensure(net_id_);
	TFingerVector::iterator itr = fingers_.begin(),
		itre = fingers_.end();
	for ( ; itr!=itre; itr++ ){
		(*itr).ensure(net_id_);
	}
}

void ChordProtocol::checkKeys()
{
	if ( !(node_->isOnline() && isActive()) ) return;

	key_timer_->resched( config.chordprot_.check_keys_intv );
	if ( !(getPredecessor().isValid()) ) return;
	if ( !(getSuccessor().isValid()) ) return;
	TChordKeyInfoList::iterator itr = key_list_.begin(),
		itre = key_list_.end(), itrb;
	while ( itr!=itre ){
		itrb = itr;
		itr++;
		ChordKeyInfo *key_info = *itrb;
		if ( !(key_info->isReady()) ) continue;
		chord_key_t key = key_info->getKey();
		if ( isSuccessorOf(key) ) continue;
		if ( isPreceding(key) ){//transfer key to successor
			ChordNode* suc_node = getSuccessorNode();
			if ( transferKeyTo( key_info, suc_node ) ){
				removeKey( itrb );
				continue;
			}
		} 
		//search for key
		int result = searchKey( key, MsgItem::KEY_TRANSFER ); 
		if ( 1==result ) key_info->setSearching(); 
	}
}

bool ChordProtocol::transferKeyTo( ChordKeyInfo *key_info, ChordNode *dst )
{
	if ( NULL==dst ) return false;
	if ( !(dst->isOnline()) ) return false;

	PeerLoger::instance().logChord( 0, "node:%u@%u transfer key:%u to node:%u@%u", node_->getChordNodeID(), node_->getNodeID(), key_info->getKey(), dst->getChordNodeID(), dst->getNodeID() );

	dst->getProtocolByID(net_id_)->addKey(key_info); 
	RELEASE_OBJ( key_info );


	return true;
}

bool ChordProtocol::transferKeyTo( chord_key_t key, node_id_t dst_id )
{
	TChordKeyInfoList::iterator itr = std::find_if( key_list_.begin(),
			key_list_.end(), __KEYINFO_KEY_PRED(key) );
	if ( key_list_.end()==itr ) return false;
	Node *dst_node = UnderTopoMgr::instance().node(dst_id);
	if ( NULL==dst_node ) return false;
	if ( transferKeyTo( *itr, static_cast<ChordNode*>(dst_node) ) ){
		removeKey( itr );
		return true;
	} else{
		(*itr)->setReady();
		return false;
	}
}

int ChordProtocol::searchKey( chord_key_t key, MsgItem::PURPOSE p, bool data )
{
	//hit immediately
	if ( isPreceding( key ) &&  (!data) )
		return 0;
	if ( isSuccessorOf( key ) ){
		if ( data ){//whether the data is required
			if ( hasKey( key ) )
				return 0;
		} else
			return 0;
	}

	ChordNodeInfo& s = closestPrecedingFinger(key);
	if ( s.isValid() ) {
		NEW_VAR_OBJECT( ChordSearchMsg, msg, ( key, ChordMsg::TARGET_SUCCESSOR, getChordNetID(), node_->getNodeID() ) );
		msg->data_ = data;
		node_->send( s.node_id_, msg );
		addPendingMsg( msg, p );
		RELEASE_OBJ( msg );
		return 1;
	} else{
		PeerLoger::instance().logError(0, "search of key:%u unreachable", key);
		return 2;//oops...
	}
}

void ChordProtocol::generateQuery( chord_key_t key )
{
	int result = searchKey( key, MsgItem::QUERY, true );
	if ( !(node_->isOnline()) ){
		assert( false );
		fprintf( stderr, "error\n" );
		return;
	}
	if ( 0==result ){
		PeerLoger::instance().logResult(0, "query %u hit:4 hops:0", key);
	} else if( 2==result ){
		PeerLoger::instance().logResult(0, "query %u hit:0 hops:0", key);
	}
}

void ChordProtocol::processMessage( ChordMsg *msg )
{
	if (!isActive()) return;
	switch( msg->getType() ){
		case ChordMsg::SEARCH_MSG:
			processSearchMsg( static_cast<ChordSearchMsg*>(msg) );
			break;
		case ChordMsg::SEARCH_RESP_MSG:
			processSearchRespMsg( static_cast<ChordSearchRespMsg*>(msg) );
			break;
		case ChordMsg::NOTIFY_MSG:
			processNotifyMsg( static_cast<ChordNotifyMsg*>(msg) );
			break;
		case ChordMsg::GET_MSG:
			processGetMsg( static_cast<ChordGetMsg*>(msg) );
			break;
		case ChordMsg::GET_RESP_MSG:
			processGetRespMsg( static_cast<ChordGetRespMsg*>(msg) );
			break;
		default:
			assert(false);
	}
}

void ChordProtocol::processSearchMsg( ChordSearchMsg *msg )
{
	assert_or_ret( NULL!=msg );
	NEW_VAR_OBJECT( ChordSearchRespMsg, rsp_msg, ( msg->key_, getChordNetID(), node_->getNodeID() ) );
	rsp_msg->msg_id_ = msg->msg_id_;

	if ( isPreceding( msg->key_ ) && 
			( (!(msg->data_)) || msg->predecessor() ) ){//hit
		if ( msg->successor() ){
			ChordNodeInfo& s = getSuccessor();
			if ( s.isValid() )rsp_msg->setHit( s.node_id_, s.chord_nid_);
			else {
				rsp_msg->setHit( node_->getNodeID(), node_->getChordNodeID() );
			} 
		} else{
			assert( !(msg->data_) );
			rsp_msg->setHit( node_->getNodeID(), node_->getChordNodeID() );
		}	

	} else if ( isSuccessorOf( msg->key_ ) ){
		if ( msg->successor() ){
			if ( msg->data_ ){//whether the data is required
				if ( hasKey( msg->key_ ) )
					rsp_msg->setHit( node_->getNodeID(), node_->getChordNodeID() );
				else 
					rsp_msg->setMiss();
			} else{
				rsp_msg->setHit( node_->getNodeID(), node_->getChordNodeID() );
			}
		} else{
			assert( !(msg->data_) );
			ChordNodeInfo& pred = getPredecessor();
			if ( pred.isValid() ){
				rsp_msg->setHit( pred.node_id_, pred.chord_nid_ );
			} else{
				rsp_msg->setHit( node_->getNodeID(), node_->getChordNodeID() );
			}
		}

	}else{//forward
		ChordNodeInfo& s = closestPrecedingFinger(msg->key_);
		if ( s.isValid() ) {
			rsp_msg->setForward( s.node_id_ );
		} else{
			rsp_msg->setMiss();
		}
	}

	node_->send( msg->getSrcID(), rsp_msg );
	RELEASE_OBJ( rsp_msg );
}

void ChordProtocol::processSearchRespMsg( ChordSearchRespMsg *msg )
{
	assert( NULL!=msg );

	TMsgItemList::iterator itr = 
		std::find_if( pending_msg_.begin(), pending_msg_.end(), 
		__MSGITEM__ID_PRED( msg->msg_id_) );
	ChordSearchMsg *search_msg = NULL;
	MsgItem msg_item;
	if ( itr!=pending_msg_.end() ){
		search_msg = static_cast<ChordSearchMsg*>((*itr).msg_);
		msg_item = *itr;
		pending_msg_.erase(itr);
	} else{
		assert(false);
		return;//we don't know what to do here
	}

	if ( ChordMsg::CODE_FORWARD==msg->code_ ){
		if ( (MsgItem::MAINTAINANCE==msg_item.purpose_) && 
				(search_msg->hops()>34) ){
			TChordKeyInfoList::iterator itr = std::find_if( key_list_.begin(), key_list_.end(), __KEYINFO_KEY_PRED( msg->key_ ) );
			if ( itr!=key_list_.end() ) (*itr)->setReady();
		} else if ( (MsgItem::QUERY==msg_item.purpose_) && 
					(config.chordprot_.search_max_hops>0) &&
					(search_msg->hops()>config.chordprot_.search_max_hops) ) 
		{
			PeerLoger::instance().logResult(0, "node:%u@%u query %u hit:0 hops:%u", 
					node_->getChordNodeID(), node_->getNodeID(), 
					search_msg->key_, search_msg->hops());
		} else{
			node_->send( msg->info_.node_id_, search_msg );
			//TODO: move the old elem to the tail of list and change its issue_time_ instead of inserting a new element
			addPendingMsg( search_msg, msg_item.purpose_ );
		}


	} else if ( ChordMsg::CODE_HIT==msg->code_ ){

		if ( MsgItem::MAINTAINANCE==msg_item.purpose_ ){
			updateFingerTable( msg );
			if ( getStatus()==CP_JOINING ){
				setReady();
				ChordGod::instance().registerToChordNet( net_id_, node_->getNodeID() );
			}
			if ( notify_ ) notify();
		} else if ( MsgItem::QUERY==msg_item.purpose_ ){//query
			PeerLoger::instance().logResult(0, "node:%u@%u query %u hit:1 hops:%u", 
					node_->getChordNodeID(), node_->getNodeID(), 
					search_msg->key_, search_msg->hops());
		} else if ( MsgItem::KEY_TRANSFER==msg_item.purpose_ ){
			transferKeyTo( msg->key_, msg->getSrcID() );
		} else {
			assert(false);//impossible
		}
		
	} else if ( ChordMsg::CODE_MISS==msg->code_ ){
		if ( MsgItem::QUERY==msg_item.purpose_ ){//query
			PeerLoger::instance().logResult(0, "node:%u@%u query %u hit:0 hops:%u", 
					node_->getChordNodeID(), node_->getNodeID(), 
					search_msg->key_, search_msg->hops());
		} else if ( MsgItem::KEY_TRANSFER==msg_item.purpose_ ){
			TChordKeyInfoList::iterator itr = std::find_if( key_list_.begin(), 
					key_list_.end(), __KEYINFO_KEY_PRED( msg->key_ ) );
			if ( itr!=key_list_.end() ) (*itr)->setReady();
		}
	}

	RELEASE_OBJ( search_msg );//release pending msg
}

void ChordProtocol::processNotifyMsg( ChordNotifyMsg *msg )
{
	assert_or_ret( NULL!=msg );
	ChordNodeInfo& pred = getPredecessor();
	if ( ( !pred.isValid() ) || 
		 ( msg->info_.isValid() && 
		   Helper::isBetween( msg->info_.chord_nid_, pred.chord_nid_, node_->getChordNodeID() ) ) ){
		setPredecessor( msg->info_.node_id_, msg->info_.chord_nid_ );
	}
}

void ChordProtocol::processGetMsg( ChordGetMsg *msg )
{
	assert_or_ret( NULL!=msg );
	NEW_VAR_OBJECT( ChordGetRespMsg, resp_msg, (net_id_, node_->getNodeID()) ); 
	resp_msg->msg_id_ = msg->msg_id_;
	if ( ChordMsg::TARGET_SUCCESSOR==msg->target_ ){
		resp_msg->info_ = getSuccessor();

	} else if ( ChordMsg::TARGET_PREDCESSOR==msg->target_ ){
		resp_msg->info_ = getPredecessor();
	}
	node_->send( msg->getSrcID(), resp_msg );
	RELEASE_OBJ( resp_msg );
}

void ChordProtocol::processGetRespMsg( ChordGetRespMsg *msg )
{
	assert_or_ret( NULL!=msg );

	TMsgItemList::iterator itr = 
		std::find_if( pending_msg_.begin(), pending_msg_.end(), 
		__MSGITEM__ID_PRED( msg->msg_id_) );
	ChordGetMsg *get_msg = NULL;
	if ( itr!=pending_msg_.end() ){
		get_msg = static_cast<ChordGetMsg*>((*itr).msg_);
		pending_msg_.erase( itr );
	} else{
		assert( false );
		//don't know what to do here
	}

	//stabilization
	ChordNodeInfo& suc_info = getSuccessor();
	if ( (ChordMsg::TARGET_PREDCESSOR==get_msg->target_) && 
			suc_info.isValid() && msg->info_.isValid() && 
			(msg->getSrcID()==suc_info.node_id_) ){
		if ( Helper::isBetween( msg->info_.chord_nid_, node_->getChordNodeID(), suc_info.chord_nid_ ) ){
			setSuccessor( msg->info_ );
			if (notify_) notify();
		} 
	}
	
	RELEASE_OBJ( get_msg );

}

bool ChordProtocol::updateFingerTable( ChordSearchRespMsg *msg )
{
	assert_or_retv( msg!=NULL, false );
	assert_or_retv( msg->info_.isValid(), false );
	if ( msg->info_.node_id_==node_->getNodeID() ) return false;

	chord_key_t key = msg->key_;
	uint32_t index = keyToFingerIndex( key );
	assert_or_retv( index<fingers_.size(), false );

	setFinger( msg->info_.node_id_, msg->info_.chord_nid_, index );
	return true;
}


void ChordProtocol::addPendingMsg( ChordMsg *msg, MsgItem::PURPOSE p /*=MsgItem::MAINTAINANCE*/ )
{
	assert( msg!=NULL );
	msg->add_reference();
	pending_msg_.push_back( MsgItem( msg, Simulator::instance().clock(), p) );

}

bool ChordProtocol::addKey( ChordKeyInfo *key_info )
{
	assert_or_retv( NULL!=key_info, false );
	key_list_.push_back( key_info );
	key_info->setReady();
	key_info->add_reference();
	number_of_key_ ++;
	return true;
}

void ChordProtocol::removeKey( TChordKeyInfoList::iterator pos )
{
	key_list_.erase( pos );
	number_of_key_--;
}

bool ChordProtocol::hasKey( chord_key_t key )
{
	TChordKeyInfoList::iterator itr = std::find_if( key_list_.begin(),
			key_list_.end(), __KEYINFO_KEY_PRED(key) );
	return ( key_list_.end()!=itr ); 
}

ChordNode* ChordProtocol::getSuccessorNode()
{
	return getFingerNode( 0 );
}

ChordNodeInfo& ChordProtocol::getSuccessor()
{
	assert( fingers_.size()>0 );
	return fingers_[0];
}

ChordNodeInfo& ChordProtocol::getPredecessor()
{
	return predecessor_;
}

ChordNode* ChordProtocol::getPredecessorNode()
{
	Node* n = UnderTopoMgr::instance().node(predecessor_.node_id_);
	if ( NULL==n ) return NULL;
	return static_cast<ChordNode*>(n);
}

void ChordProtocol::setSuccessor(node_id_t n_id, chord_node_id_t c_nid) 
{ 
	return setFinger( n_id, c_nid, 0 );
}

void ChordProtocol::setSuccessor(ChordNodeInfo& info)
{
	setFinger( info.node_id_, info.chord_nid_, 0 );
}

bool ChordProtocol::setPredecessor(node_id_t n_id, chord_node_id_t c_nid)
{
	assert_or_retv( n_id!=invalid_node_id, false );
	if ( n_id==node_->getNodeID() ) return false;

	//during merging, a moved out peer may send a notify message before
	//it leave. If the nofity mesage arrives after its leave, this may 
	//rise an error during the merging since the structure is incorrect
	FingerItem tmp;
	tmp.node_id_ = n_id;
	tmp.chord_nid_ = c_nid;
	tmp.ensure(net_id_);
	if ( !tmp.isValid() ) return false;
		

	predecessor_.node_id_ = n_id;
	predecessor_.chord_nid_ = c_nid;
	predecessor_.last_update_ = Simulator::instance().clock();

	if ( (!getSuccessor().isValid()) && first_node_start_ ){
		first_node_start_ = false;
		setSuccessor( n_id, c_nid );
	}
}

ChordNode* ChordProtocol::getFingerNode(uint32_t index)
{
	assert_or_retv( fingers_.size()>index, NULL );
	Node* n = UnderTopoMgr::instance().node(fingers_[index].node_id_);
	if ( NULL==n ) return NULL;
	return static_cast<ChordNode*>(n);
}

FingerItem& ChordProtocol::getFinger( uint32_t index ) 
{ 
	assert( fingers_.size()>index );
	return fingers_[index]; 
}

void ChordProtocol::setFinger(node_id_t n_id, chord_node_id_t c_nid, uint32_t index)
{
	assert_or_ret( fingers_.size()>index );
	if ( n_id==node_->getNodeID() ) return ;
#if 1
	if ( 0==index ){
		FingerItem& old_info = fingers_[index];
		old_info.ensure(net_id_);
		if ( old_info.isValid() && (!Helper::isBetween( c_nid, node_->getChordNodeID(), old_info.chord_nid_ )) )
			return;
		//if ( suc.isValid() && (!Helper::isBetweenLt( c_nid, old_info.start_, old_info. )) )
	}
#endif

	//during merging, a moved out peer may send a notify message before
	//it leave. If the nofity mesage arrives after its leave, this may 
	//rise an error during the merging since the structure is incorrect
	FingerItem tmp;
	tmp.node_id_ = n_id;
	tmp.chord_nid_ = c_nid;
	tmp.ensure(net_id_);
	if ( !tmp.isValid() ) return ;
		

	fingers_[index].node_id_ = n_id; 
	fingers_[index].chord_nid_ = c_nid;
	fingers_[index].last_update_ = Simulator::instance().clock();
	if ( 0==index ) notify_ = true;
}

uint32_t ChordProtocol::determineFingerTableSize()
{
	//uint32_t s = (uint32_t) std::ceil( Helper::log( 
			//ChordGod::instance().getKeySpaceSize(), 2 ) );
	//asset( 32==s );//currently we use 2^32 as key space
	return 32;
}

bool ChordProtocol::isPreceding( chord_key_t k )
{
	if (config.chordgod_.initial_chord_size <= 0) {
		assert_or_retv( getStatus()==CP_READY, false );
	}
	ChordNodeInfo& successor = getSuccessor();
	if ( successor.isValid() ){
		return Helper::isBetweenGt( k, node_->getChordNodeID(), successor.chord_nid_ );
	}
	else {
		return ( NULL== ChordGod::instance().getNodeInChordNet( net_id_, node_->getNodeID()) );
	}
}

bool ChordProtocol::isSuccessorOf( chord_key_t k)
{
	if (config.chordgod_.initial_chord_size <= 0) {
		assert_or_retv( getStatus()==CP_READY, false );
	}
	ChordNodeInfo& pred = getPredecessor();
	if ( pred.isValid() ){
		return Helper::isBetweenGt( k, pred.chord_nid_, node_->getChordNodeID() );
	} else{
		return ( NULL== ChordGod::instance().getNodeInChordNet( net_id_, node_->getNodeID()) );
	}
}

uint32_t ChordProtocol::keyToFingerIndex( chord_key_t key )
{
	return keyToFingerIndex( node_->getChordNodeID(), key );
}

void ChordProtocol::setReady()
{
	setStatus(CP_READY);
	//rt_timer_->resched( config.chordprot_.stabilize_intv );
	//key_timer_->resched( config.chordprot_.check_keys_intv );
	rt_timer_->resched( 0.0 );
	key_timer_->resched( 0.0 );

	if ( isMerging() ) 
		onMergeDone();
}

void ChordProtocol::notify()
{
	ChordNodeInfo& suc_info = getSuccessor();
	if ( suc_info.isValid() ){
		NEW_VAR_OBJECT( ChordNotifyMsg, ntf_msg, (node_->getChordNodeID(), node_->getNodeID(), net_id_) );
		node_->send( suc_info.node_id_, ntf_msg );
		RELEASE_OBJ( ntf_msg );
	}
	notify_ = false;
}

FingerItem& ChordProtocol::closestPrecedingFinger( chord_key_t key )
{
	TFingerVector::reverse_iterator ritr = fingers_.rbegin(),
		ritre = fingers_.rend();
	for ( ; ritr!=ritre; ritr++ ){
		FingerItem& f = *ritr;
		f.ensure(net_id_);
		if ( f.isValid() && 
			 Helper::isBetween( f.chord_nid_, node_->getChordNodeID(), key ) ){
			return f;
		}
	}
	//return fingers_[0];
	return firstValidFinger();
}

FingerItem& ChordProtocol::firstValidFinger()
{
	TFingerVector::iterator itr = fingers_.begin(),
		itre = fingers_.end();
	for ( ; itr!=itre; itr++ ){
		if ( itr->isValid() )
			return *itr;
	}

	return fingers_[0];
}

void ChordProtocol::getKeysFromGod()
{
	ChordNetInfo& net_info = ChordGod::instance().getNetInfo( net_id_ );
	if( net_info.key_tuples_==NULL ){
		assert( isMerging() );
		return;
	}


	for ( int i=0; i<net_info.number_of_keys_; i++ ){
		NEW_VAR_OBJECT( ChordKeyInfo, key_info, (&(net_info.key_tuples_[i]), 0) );
		key_list_.push_back( key_info );
		number_of_key_++;
	}
}

void ChordProtocol::onMergeDone()
{
	assert( isMerging() );
	chord_net_id_t old_net_id = getOldNetID();

	ChordProtocol *p_old = node_->getProtocolByID( old_net_id );
	assert_or_ret( p_old!=NULL );
	p_old->moveOut( net_id_ );
}

void ChordProtocol::moveOut( chord_net_id_t new_net_id )
{
	setStatus( CP_MOVED_OUT );
	new_net_id_ = new_net_id;
	PeerLoger::instance().logDebug( 0, "node:%u@%u moves from:%u to:%u", node_->getChordNodeID(), node_->getNodeID(), net_id_, new_net_id );
	ChordGod::instance().unregisterToChordNet( net_id_, node_->getNodeID() );
	fingers_[0].ensure( net_id_ );
	predecessor_.ensure( net_id_ );
	ChordNode *s_n = getSuccessorNode();
	ChordNode *p_n = getPredecessorNode();
	if ( (NULL!=s_n) && (NULL==p_n)  ){
		fprintf( stderr, "\nnode:%u:%u moving out failed: no successor.\n", node_->getChordNodeID(), node_->getNodeID() );
		return ;
	}
	if ( NULL==s_n )
	{
		fprintf( stderr, "\nnode:%u:%u moving out finished: no successor.\n", node_->getChordNodeID(), node_->getNodeID() );
		return ;
	}
	//TODO: send a message to successor
	ChordProtocol *s_p = s_n->getProtocolByID( net_id_ );
	s_p->setPredecessor( p_n->getNodeID(), p_n->getChordNodeID() );
	ChordProtocol *p_p = p_n->getProtocolByID( net_id_ );
	p_p->fingers_[0].node_id_ = s_n->getNodeID();
	p_p->fingers_[0].chord_nid_ = s_n->getChordNodeID();
	s_n->merge( net_id_, new_net_id_ );

}

/** 
 * @brief This method determins which finger item is responssible for a
 * specific key. node ID is needed.
 * 
 * @param base is the node ID
 * @param key denotes the key whose finger item is to be calculated.
 * 
 * @return the index of the finger which is responssible for key
 */
uint32_t ChordProtocol::keyToFingerIndex( chord_key_t base, chord_key_t key )
{
	assert_or_retv( (key!=base), invalid_chord_node_id );
	double diff = (double)Helper::round_distance( base, key );
	uint32_t i = (uint32_t) (Helper::log( diff, 2 )+epsilon);
	return i;
}

/** 
 * @brief calculate the range of corresponding finger table item
 *
 * At a node whose ID is n, the range of the ith item of the finger 
 * table is [ n+2^i, n+2^(i+1) ), where i = 0, 1, 2...
 * Thus, when this method returns, the calculated range is:
 * 		[left, right)
 * 
 * @param base is the node ID
 * @param index denotes which finger item to calculate
 * @param left stores the left boundary of the range when return 
 * @param right stores the right boundary of the range when return  
 */
void ChordProtocol::fingerRange( chord_key_t base, uint32_t index, chord_key_t& left, chord_key_t& right )
{
	left = base + (chord_key_t)std::pow( (double)2, (double)index );
	right = base + (chord_key_t)std::pow( (double)2, (double)(index+1) );
}

void ChordProtocol::fingerString( std::string& s )
{
	std::stringstream ss;
	ss<<"fingers(";
	TFingerVector::iterator itr = fingers_.begin(),
		itre = fingers_.end();
	for ( int i=0; itr!=itre; itr++,i++ ){
		FingerItem& f = *itr;
		if ( f.isValid() )
			ss<<i<<":"<<f.chord_nid_<<"@"<<f.node_id_<<" ";
		else
			ss<<i<<":null ";
	}
	ss<<")";
	s = ss.str();
}

void ChordProtocol::keyString( std::string& s )
{
	std::stringstream ss;
	ss<<" key(total:"<<getNumberOfKey()<<" ";
	TChordKeyInfoList::iterator itr = key_list_.begin(),
		itre = key_list_.end();
	for ( ; itr!=itre; itr++ ){
		ChordKeyInfo *info = *itr;
		ss<<info->getKey()<<" ";
	}
	ss<<")";
	s = ss.str();
}



